package main

import (
	"fmt"
	"mapreduce"
	"strconv"
	"strings"
	"unicode"
)

const (
	jobName = "job_213432431243"
	masterAddress = "localhost:8001"
	worker1Address = "localhost:9001"
)

func main() {
	//有多少个文件就有多少个map任务,
	//逐个执行map任务, 再逐个执行reduce任务
	files := []string{"F:/tmp/mr/input0.txt", "F:/tmp/mr/input1.txt","F:/tmp/mr/input2.txt",
	"F:/tmp/mr/input3.txt", "F:/tmp/mr/input4.txt", "F:/tmp/mr/input5.txt"}
	mapreduce.Sequential(jobName,files, 2,mapF, reduceF)

	//master := mapreduce.Distributed(jobName, files, 2, masterAddress)
	//mapreduce.RunWorker(masterAddress, worker1Address, mapF, reduceF, -1)
	//fmt.Println(*master)
}

func mapF(file string, contents string) []mapreduce.KeyValue{
	bytes := []byte(contents)
	isWordEnd := false
	var words []string
	var wordBytes []byte
	var resultPairs []mapreduce.KeyValue

	for i, b := range bytes {
		isLetter := unicode.IsLetter(rune(b))
		//单词开始
		if isLetter {
			isWordEnd = false
			wordBytes = append(wordBytes, b)
		} else {
			isWordEnd = true
		}

		if i == len(bytes)-1 {
			isWordEnd = true
		}

		if len(wordBytes) != 0 && isWordEnd {
			word := string(wordBytes)
			word = strings.ToLower(word)
			words = append(words, word)
			kv := mapreduce.KeyValue{word, "1"}
			resultPairs = append(resultPairs, kv)
			wordBytes = wordBytes[:0]
			isWordEnd = true
		}
	}
	return resultPairs
}

func reduceF(key string, values []string) string {
	return fmt.Sprintf("%s %s", key, strconv.Itoa(len(values)))
}
